package NIOCommunication;

import java.io.IOException;
import java.nio.channels.CancelledKeyException;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.ClosedSelectorException;
import java.nio.channels.SelectableChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.ConcurrentModificationException;
import java.util.Iterator;
import java.util.Vector;

public class NIOProcessor
{
	private Selector mSelector;
	private Vector<NIOProcessorCommand> mCommands;
	private NIOProcessorThread mThread;
        private Vector<NIOSocketContext> mPossibleTimeouts;
	
	public NIOProcessor()
	{
		try
		{
			mSelector = Selector.open();
		}
		catch (IOException e)
		{
			// TODO Auto-generated catch block
			e.printStackTrace();
		}
		
		mCommands = new Vector<NIOProcessorCommand>();
                mPossibleTimeouts = new Vector<NIOSocketContext>();
	}
        
        public boolean IsClosed()
        {
            return mSelector == null || !mSelector.isOpen();
        }
	
	public void RegisterContextOps(NIOContext context, int ops) throws ClosedChannelException
	{
		SelectableChannel channel = context.GetChannel();
		if(channel == null)
		{
			return;
		}
		
		channel.register(mSelector, ops, context);
		//System.out.println("Context " + context.GetContextID() +  " registered " + ((ops & SelectionKey.OP_CONNECT) != 0 ? "Connect ":"")  + ((ops & SelectionKey.OP_ACCEPT) != 0 ? "Accept ":"") + ((ops & SelectionKey.OP_READ) != 0 ? "Read ":"") + ((ops & SelectionKey.OP_WRITE) != 0 ? "Write ":""));
		
                synchronized(mSelector)
                {
                    if(mThread == null)
                    {
                    	mThread = new NIOProcessorThread(this, mSelector, mCommands, mPossibleTimeouts);
                    }
                }
	}
        
        public void EnsureThreadIsRunning()
        {
                synchronized(mSelector)
                {
                    if(mThread == null)
                    {
                    	mThread = new NIOProcessorThread(this, mSelector, mCommands, mPossibleTimeouts);
                    }
                }
        }
        
        public void CloseAllExcept(NIOContext toSpare)
        {
            synchronized(mSelector)
            {
                Iterator<SelectionKey> it = mSelector.keys().iterator();
                while(it.hasNext())
                {
                    NIOContext context = (NIOContext)it.next().attachment();
                    if(context != toSpare)
                    {
                        context.Close();
                    }
                }
            }
        }
        
        public void Kill()
        {
            try
            {
                synchronized(mSelector)
                {
                    CloseAll();
                    mSelector.select();
                    mSelector.close();
                    mSelector = null;
                }
            }
            catch(Exception e)
            {
                e.printStackTrace();
            }
            /*if(mThread != null)
            {
                mThread.mRunning = false;
                while(mThread != null);
                CloseAll();
                try
                {
                    mSelector.select();
                }
                catch(Exception e)
                {
                    e.printStackTrace();
                }
                mSelector.close();
            }*/
        }
        
        public void CloseAll()
        {
            synchronized(mSelector)
            {
                Iterator<SelectionKey> it = mSelector.keys().iterator();
                while(it.hasNext())
                {
                    NIOContext context = (NIOContext)it.next().attachment();
                    context.Close();
                }
            }
        }
        
        public void SetConnectionTimeout(NIOSocketContext context)
        {
            mPossibleTimeouts.add(context);
        }
        
        public void ClearConnectionTimeout(NIOSocketContext context)
        {
            mPossibleTimeouts.remove(context);
        }
	
	public void QueueCommand(NIOProcessorCommand command)
	{
		mCommands.add(command);
		
                synchronized(mSelector)
                {
                    if(mThread == null)
                    {
                    	mThread = new NIOProcessorThread(this, mSelector, mCommands, mPossibleTimeouts);
                    }
                }
	}
	
	private void NotifyThreadQuit()
	{
		System.out.println("NIOProcessorThread quit");
		mThread = null;
	}
	
	private class NIOProcessorThread extends Thread
	{
		private NIOProcessor mProcessor;
		private Selector mSelector;
		private boolean mRunning;
		private Vector<NIOProcessorCommand> mCommands;
                private Vector<NIOSocketContext> mPossibleTimeouts;
		
		public NIOProcessorThread(NIOProcessor processor, Selector selector, Vector<NIOProcessorCommand> commands, Vector<NIOSocketContext> possibleTimeouts)
		{
			mProcessor = processor;
			mSelector = selector;
			mCommands = commands;
                        mPossibleTimeouts = possibleTimeouts;
			this.start();
		}
	
		public void run()
		{
			System.out.println("NIOProcessorThread running");
			mRunning = true;
			while(mRunning)
			{
                            //System.out.println("Loop");
                            
                            DoSelect();
			
                            DoDispatch();
                            
                            CheckForTimeouts();
                            
                            DoCommands();
					
                            if(ShouldExit())
                            {
                                break;
                            }   
			}
                    mProcessor.NotifyThreadQuit();
                        
		}
        
        private void DoSelect()
        {
            try
            {
                synchronized(mSelector)
                {
                    mSelector.select(100); // timeout ever 100ms in case some non-selector event happened
                }
            }	
            catch(IOException e)
            {
                // TODO Auto-generated catch block
                e.printStackTrace();
            }
            catch(ClosedSelectorException e)
            {
                // TODO Auto-generated catch block
		e.printStackTrace();
            }
        }
        
        private void DoDispatch()
        {
            Iterator<SelectionKey> it = mSelector.selectedKeys().iterator();
            while(it.hasNext())
            {
		SelectionKey key = it.next();
		if(key.isValid())
		{
                    try
                    {
                        NIOContext context = (NIOContext)key.attachment();
			context.PerformActions(key);
                    }
                    catch(CancelledKeyException e)
                    {
			//e.printStackTrace();
                    }
                    catch(Exception e)
                    {
                        e.printStackTrace();
                    }	
		}
			
		it.remove();
            }
        }
        
        private void CheckForTimeouts()
        {
            // check for connection timeouts
            try
            {
                synchronized (mSelector){
                synchronized (mPossibleTimeouts)
                {
                    Iterator<NIOSocketContext> contextIt = mPossibleTimeouts.iterator();
                    while (contextIt.hasNext())
                    {
                        NIOSocketContext context = contextIt.next();
                        if (context.IsTimeoutElapsed())
                        {
                            //System.out.println("Connection timed out");
                            //context.Close();
                            mCommands.add(new NIOProcessorCloseContextCommand(context, true));
                            contextIt.remove();
                        }
                    }
                }}
            }
            catch (ConcurrentModificationException e)
            {
                e.printStackTrace();
            }
        }
        
        private void DoCommands()
        {
            // now process any pending commands
            synchronized (mSelector){
            synchronized (mCommands)
            {
                Iterator<NIOProcessorCommand> commandIt = mCommands.iterator();
                while (commandIt.hasNext())
                {
                    try
                    {
                        commandIt.next().Execute(mProcessor);
                    }
                    catch (Exception e)
                    {
                        e.printStackTrace();
                    }
                    commandIt.remove();
                }
            }}
        }
        
        int mShouldExitCount = 100;
        private boolean ShouldExit()
        {
            synchronized (mSelector)
            {
                // check if all the keys have been closed
                try
                {
                    if (mSelector.keys().size() == 0)
                    {
                        --mShouldExitCount;
                        if(mShouldExitCount <= 0)
                        {
                            mRunning = false;

                            System.out.println("Selector is empty.  Selector thread exiting.");
                            return true;
                        }
                    }
                }
                catch(ClosedSelectorException e)
                {
                    return true;
                }
            }
            
            return false;
        }
    }
}
